package com.xxwu.messagequeue.mqserver;

import com.xxwu.messagequeue.common.Consumer;
import com.xxwu.messagequeue.common.ConsumerEnv;
import com.xxwu.messagequeue.common.MqException;
import com.xxwu.messagequeue.mqserver.core.*;
import com.xxwu.messagequeue.mqserver.datacenter.DiskDataCenter;
import com.xxwu.messagequeue.mqserver.datacenter.MemoryDataCenter;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 创建虚拟主机用于对硬盘和内存上的数据操作进行整合
 * 锁的粒度较大，但创建和删除虽然该类中这些核心部件的操作都不频繁
 * 并且 synchronized 是偏向锁，在没有锁竞争时不会加锁
 * 所以并没有造成较大的效率问题
 */
@Data
@Slf4j
public class VirtualHost {
    // 这里为了实现交换机与虚拟主机之间的对应关系
    // 直接使用 exchangeName = virtualHostName + exchangeName 来表示对应关系
    private String virtualHostName; // 主机的表示
    private MemoryDataCenter memoryDataCenter = new MemoryDataCenter();
    private DiskDataCenter diskDataCenter = new DiskDataCenter();
    private Router router = new Router();
    private ConsumerManager consumerManager = new ConsumerManager(this);;

    // 线程安全的锁对象
    // 此处加上 final 就没有标黄，因为 idea 为防止这个锁对象后续被指向其他对象
    private final Object exchangeLocker = new Object();
    private final Object queueLocker = new Object();

    // 构造方法中实现硬盘存储的初始化以及内存的恢复
    public VirtualHost(String virtualHostName) {
        this.virtualHostName = virtualHostName;
        // 对 DiskDataCenter 进行初始化
        diskDataCenter.init();
        // 还需要将硬盘中的数据恢复到内存中
        // 设定这个类为方法调用的最上层，在这里不再对异常向上抛出
        // 而是将异常捕获进行处理，否则若这里出现异常程序将会直接崩溃
        try {
            memoryDataCenter.recovery(diskDataCenter);
        } catch (IOException | MqException | ClassNotFoundException e) {
            e.printStackTrace();
            log.error("[VirtualHost],数据恢复失败！");
        }
    }

    public boolean exchangeDeclare(String exchangeName, ExchangeType type,
                               Boolean durable, Boolean autoDelete, HashMap<String, Object> argument){
        exchangeName = virtualHostName + exchangeName;

        try{
            synchronized (exchangeLocker){
                // 判断是否已经存在该交换机
                Exchange existsExchange = memoryDataCenter.getExchange(exchangeName);
                if(existsExchange != null){
                    log.info("交换机已经存在！");
                    return true;
                }
                Exchange exchange = new Exchange();
                exchange.setName(exchangeName);
                exchange.setType(type);
                exchange.setDurable(durable);
                exchange.setAutoDelete(autoDelete);
                exchange.setArgument(argument);
                // 先写入硬盘，再写入内存，若硬盘写入失败那么则不能再写入内存
                // 将交换机写硬盘
                if(durable){
                    diskDataCenter.insertExchange(exchange);
                }
                // 写入内存
                memoryDataCenter.insertExchange(exchange);
                log.info("[VirtualHost],交换机创建成功！");
                return true;
            }
        }catch(Exception e){
            e.printStackTrace();
            log.error("[VirtualHost]，插入交换机失败！");
            return false;
        }
    }
    public boolean exchangeDelete(String exchangeName){
        exchangeName = virtualHostName + exchangeName;
        try{
            synchronized (exchangeLocker){
                // 判断交换机是否存在
                Exchange exchange = memoryDataCenter.getExchange(exchangeName);
                if(exchange == null){
                    throw new MqException("[VirtualHost],交换机不存在无法删除！exchangeName=" + exchangeName);
                }
                if(exchange.isDurable()){
                    diskDataCenter.deleteExchange(exchangeName);
                }
                memoryDataCenter.deleteExchange(exchangeName);
                return true;
            }
        }catch(Exception e){
            e.printStackTrace();
            log.error("[VirtualHost],交换机删除失败！exchangeName={}", exchangeName);
            return false;
        }
    }
    public boolean queueDeclare(String queueName, Boolean durable, Boolean autoDelete
            , Boolean exclusive, HashMap<String, Object> argument){
        queueName = virtualHostName + queueName;
        try{
            synchronized (queueLocker){
                MSGQueue existsQueue = memoryDataCenter.getQueue(queueName);
                if(existsQueue != null){
                    log.info("[VirtualHost],队列已存在 queueName={}", queueName);
                    return true;
                }
                MSGQueue queue = new MSGQueue();
                queue.setName(queueName);
                queue.setDurable(durable);
                queue.setAutoDelete(autoDelete);
                queue.setExclusive(exclusive);
                queue.setArgument(argument);
                if(durable){
                    diskDataCenter.insertMSGQueue(queue);
                }
                memoryDataCenter.insertQueue(queue);
                log.info("[VirtualHost],队列插入成功 queueName={}", queueName);
                return true;
            }
        }catch (Exception e){
            e.printStackTrace();
            log.error("[VirtualHost],插入队列失败！queueName={}", queueName);
            return false;
        }
    }
    public boolean queueDelete(String queueName){
        queueName = virtualHostName + queueName;
        try{
            synchronized (queueLocker){
                MSGQueue queue = memoryDataCenter.getQueue(queueName);
                if(queue == null){
                    throw new MqException("[VirtualHost],队列不存在删除失败 queueName=" + queueName);
                }
                if(queue.isDurable()){
                    diskDataCenter.deleteMSGQueue(queueName);
                }
                memoryDataCenter.deleteQueue(queueName);
                return true;
            }
        }catch (Exception e){
            e.printStackTrace();
            log.error("[VirtualHost],删除队列失败 queueName={}", queueName);
            return false;
        }
    }
    public boolean queueBind(String exchangeName, String queueName, String bindingKey){
        exchangeName = virtualHostName + exchangeName;
        queueName = virtualHostName + queueName;
        try{
            // 在这里要保证加锁顺序与删除操作相同，避免产生循环等待
            synchronized (exchangeLocker){
                synchronized (queueLocker){
                    Binding existsBinding = memoryDataCenter.getBinding(exchangeName, queueName);
                    if(existsBinding != null){
                        log.info("[VirtualHost],绑定已存在" +
                                "exchangeName={}, queueName={}", exchangeName, queueName);
                        return true;
                    }
                    Exchange exchange = memoryDataCenter.getExchange(exchangeName);
                    if(exchange == null){
                        throw new MqException("[VirtualHost],交换机不存在 exchangeName=" + exchangeName);
                    }
                    MSGQueue queue = memoryDataCenter.getQueue(queueName);
                    if(queue == null){
                        throw new MqException("[VirtualHost],队列不存在 queueName=" + queueName);
                    }
                    if(!router.checkBindingKey(bindingKey)){
                        throw new MqException("[VirtualHost],bindingKey不合法 bindingKey=" + bindingKey);
                    }

                    Binding binding = new Binding();
                    binding.setBindingKey(bindingKey);
                    binding.setExchangeName(exchangeName);
                    binding.setQueueName(queueName);
                    if(exchange.isDurable() && queue.isDurable()){
                        diskDataCenter.insertBinding(binding);
                    }
                    memoryDataCenter.insertBinding(binding);
                    log.info("[VirtualHost],绑定创建成功！exchangeName={}, queueName={}", exchangeName, queueName);
                    return true;
                }
            }
        } catch (Exception e){
            e.printStackTrace();
            log.error("[VirtualHost],绑定创建失败 " +
                    "exchangeName={}, queueName={}", exchangeName, queueName);
            return false;
        }
    }

    public boolean queueUnbind(String exchangeName, String queueName){
        exchangeName = virtualHostName + exchangeName;
        queueName = virtualHostName + queueName;
        try{
            synchronized (exchangeLocker){
                synchronized (queueLocker){
                    Binding binding = memoryDataCenter.getBinding(exchangeName, queueName);
                    if(binding == null){
                        throw new MqException("[VirtualHost],绑定不存在 " +
                                "exchangeName=" + exchangeName + ", queueName=" + queueName);
                    }
                    // 这里不用再判断队列和交换机是否存在，正常流程是：先删除绑定再删除交换机和队列
                    // 使用判断后进行删除，倘若出现了类似用户先删除队列/交换机才删除绑定的情况，该绑定就无法删除了
                    // 也不用判断是否持久化，若硬盘中不存在，直接删除并不会抛出异常导致程序崩溃
                    diskDataCenter.deleteBinding(binding);
                    memoryDataCenter.deleteBinding(exchangeName, queueName);
                    return true;
                }
            }
        }catch (Exception e){
            e.printStackTrace();
            log.info("[VirtualHost],绑定删除失败 " +
                    "exchangeName={}, queueName={}", exchangeName, queueName);
            return false;
        }
    }

    // 发送消息到指定的交换机->队列中
    public boolean basicPublish(String exchangeName, String routingKey, BasicProperties basicProperties, byte[] body){
        try{
            // 1 转换交换机的名字
            exchangeName = virtualHostName + exchangeName;
            // 2 查看交换机是否存在
            Exchange exchange = memoryDataCenter.getExchange(exchangeName);
            if(exchange == null){
                throw new MqException("[VirtualHost],要转发的交换机不存在！exchangeName=" + exchangeName);
            }
            // 3 检验 routingKey 是否合法
            if(!router.checkRoutingKey(routingKey)){
                throw new MqException("[VirtualHost],routingKey 不合法！ routingKey=" + routingKey);
            }
            // 4 根据交换机类型进行转发
            // 若是直接交换机 那么 routingKey 就是队列的名字，直接转发即可
            if(exchange.getType() == ExchangeType.Direct){
                // 1) 转换队列名
                String queueName = virtualHostName + routingKey;
                // 2) 查看队列是否存在
                MSGQueue  queue = memoryDataCenter.getQueue(queueName);
                if(queue == null){
                    throw new MqException("[VirtualHost],要转发的队列不存在 queueName=" + queueName);
                }
                // 3) 构造消息进行转发
                Message message = Message.createMessageWithId(routingKey,basicProperties,body);
                sendMessage(queue, message);
            } else {
                // 如果是 Fanout 和 Topic 则要根据绑定关系进行转发
                // 由于 Fanout 和 Topic 的转发规则很相似这里合并写
                // 1) 根据绑定关系查找队列
                ConcurrentHashMap<String, Binding> bindsMap = memoryDataCenter.getBindings(exchangeName);
                if(bindsMap == null){
                    throw new MqException("[VirtualHost],该交换机无绑定关系 exchangeName=" + exchangeName);
                }
                // 2) 遍历根据 exchangeName 查找到的对应队列绑定集合
                for(Map.Entry<String, Binding> entry : bindsMap.entrySet()){
                    Binding binding = entry.getValue();
                    // 3 查看绑定的队列是否存在
                    MSGQueue queue = memoryDataCenter.getQueue(binding.getQueueName());
                    if(queue == null){
                        // 这里就不用抛出异常了，当前队列不存在时就继续查找知道找到队列存在的绑定关系
                        log.info("[VirtualHost],当前队列不存在");
                        continue;
                    }
                    // 如果是 Fanout 就转发给所有绑定的队列
                    // 如果是 Topic 则要看 routingKey 和 bindingKey 的对应关系进行转发
                    // 4 构造消息
                    Message message = Message.createMessageWithId(routingKey,basicProperties,body);
                    if(!router.route(exchange.getType(), binding, message)){
                        continue;
                    }
                    sendMessage(queue, message);
                }
            }
            return true;
        }catch (Exception e){
            e.printStackTrace();
            log.error("[VirtualHost],转发消息失败！");
            return false;
        }
    }

    private void sendMessage(MSGQueue queue, Message message) throws IOException, MqException {
        // 判断消息是否持久化
        if(message.getBasicProperties().getDeliveryMode() == 2){
            diskDataCenter.addMessage(queue, message);
        }
        memoryDataCenter.sendMessageQueue(queue.getName(), message);
        memoryDataCenter.insertMessage(message);
        //  此处还要补充一个逻辑，通知消费者可以取消息了
        consumerManager.addMessage(queue.getName());
    }

    // 消费者对消息进行订阅
    // 其实就是消费者调用该方法，传入的队列参数会将消费者添加到对应的队列中专门保存消费者的list中
    // 后续通过轮询的方法将消息发送给消费者进行消费
    public boolean basicConsumer(String consumerTag, String queueName, boolean autoAck, Consumer consumer){
        // 取出队列进行验证
        // 将队列名进行转化
        queueName = virtualHostName + queueName;

        try{
            MSGQueue queue = memoryDataCenter.getQueue(queueName);
            if(queue == null){
                throw new MqException("要订阅的队列不存在！");
            }
            // 将消费者添加到队列
            consumerManager.addConsumer(consumerTag,queueName, autoAck, consumer);
            log.info("消息订阅成功！");
            return true;
        } catch(Exception e){
            e.printStackTrace();
            log.error("订阅消息失败！");
            return false;
        }
    }

    // 消息的手动应答
    public boolean basicAck(String queueName, String messageId){
        try{
            // 检验队列是否存在
            MSGQueue queue = memoryDataCenter.getQueue(queueName);
            if(queue == null){
                throw new MqException("队列不存在！");
            }
            // 检验消息是否存在
            Message message = memoryDataCenter.getMessage(messageId);
            if(message == null){
                throw new MqException("消息不存在！");
            }
            // 删除硬盘上的数据
            if(message.getDeliveryMode() == 2){
                diskDataCenter.deleteMessage(queue, message);
            }
            // 删除内存上的数据
            memoryDataCenter.deleteMessage(messageId);
            memoryDataCenter.deleteMessageQueueWaitAck(queueName, message);
            return true;
        } catch (Exception e){
            e.printStackTrace();
            log.error("消息手动应答失败！");
            return false;
        }
    }
}
